package com.lonni.rpc.core.socket.client;

import com.lonni.rpc.core.loadbalance.BaseLoadbalance;
import com.lonni.rpc.core.model.RpcConfig;
import com.lonni.rpc.core.model.RpcRequest;
import com.lonni.rpc.core.model.RpcResponse;
import com.lonni.rpc.core.model.ServiceAddress;
import com.lonni.rpc.core.register.ServiceDiscovery;
import com.lonni.rpc.core.register.model.Service;
import com.lonni.rpc.core.register.model.ServiceURL;
import com.lonni.rpc.core.socket.codec.RpcProtocol;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.Promise;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.checkerframework.checker.units.qual.A;
import org.omg.PortableServer.POA;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.io.Serializable;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

/**
 * 客户端具体发送类
 *
 * @author: Lonni
 * @date: 2024/7/11 0011 10:17
 */
@Component
@Slf4j
public class NettyTransport implements Serializable {

    /**
     * 负载均衡器
     */
    @Autowired
    private BaseLoadbalance loadbalance;

    @Autowired
    private ServiceDiscovery discovery;

    @Autowired
    private RpcConfig rpcConfig;

    /**
     * 发送数据
     * 思路:
     * 因为netty是异步的,所以在发送之前需要声明一个CompletableFuture接受异步结果
     * 将 rpcRequest中的请求id和当前的future绑定
     * 当netty客户端chanenlRead0方法执行后,更具返回的requestId找到保存的future,然后设置结果
     *
     * @param protocol
     * @return   CompletableFuture<RpcProtocol<RpcResponse>>
     */
    @SneakyThrows
    public Object transport(RpcProtocol<RpcRequest> protocol) {
        //修改序列化方式
        log.info("[{}]开始执行rpc客户端逻辑","transport");
        protocol.getHeader().setSerializationType(Integer.parseInt(rpcConfig.getSerializable()));
        // 从注册中心拿到地址
        List<ServiceURL> urls = discovery.getService(Service.builder()
                .interfaceName(protocol.getBody().getInterfaceName())
                .group(protocol.getBody().getGroup())
                .version(protocol.getBody().getVersion())
                .build());
        log.info("[{}]构建服务信息载体完成","transport");
        if (CollectionUtils.isEmpty(urls)) {
            log.error("发送数据失败,未找到服务对应地址");
            throw new RuntimeException("ServiceURL is null ");
        }
        InetSocketAddress address = null;
        if (urls.size() == 1) {
            address = new InetSocketAddress(urls.get(0).getIp(), urls.get(0).getPort());
        } else {
            //负载均衡器
            List<ServiceAddress> serviceAddresses = urls.stream().map(p -> {
                return ServiceAddress.builder()
                        .ip(p.getIp())
                        .port(p.getPort())
                        .build();
            }).collect(Collectors.toList());
            ServiceAddress serviceAddress = loadbalance.getServiceAddress(serviceAddresses, protocol.getBody());
            address = new InetSocketAddress(serviceAddress.getIp(), serviceAddress.getPort());
        }
        log.info("[{}]获取服务地址完成,链接地址:{}","transport",address.toString());

        log.info("[{}]开始发送数据","transport");
        CompletableFuture<RpcProtocol<RpcResponse>> resultFuture = new CompletableFuture<>();
        log.info("[{}]当前jvm-netty客户端链接个数:{}","transport", NettyClientContext.me.getChannelCount());
        Channel channel = RpcNettyClient.client.getChannel(address);
        Promise<RpcProtocol<RpcResponse>> promise=new DefaultPromise<>(channel.eventLoop());
        if (channel.isActive()) {
            log.info("[{}]开始发送数据","transport");
            NettyClientContext.me.putFuture(protocol.getBody().getRequestId().toString(), resultFuture);
            channel.writeAndFlush(protocol).addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()) {
                    log.info("[{}]发送数据完成 :{}", "transport",protocol);
                } else {
                    //发送失败
                    future.channel().close();
                    resultFuture.completeExceptionally(future.cause());
                    log.error("发送数据失败:", future.cause());
                }
            });
        } else {
            throw new IllegalStateException();
        }

        return resultFuture;
    }


}



